﻿using Microsoft.Extensions.Logging;
using System;
using System.Buffers;
using System.Diagnostics;
using System.IO.Pipelines;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

namespace Atomic
{
    public class SocketConnection : TransportConnection
    {
        private static readonly int MinAllocBufferSize = SlabMemoryPool.BlockSize * 2;
        private static readonly bool IsWindows = RuntimeInformation.IsOSPlatform(OSPlatform.Windows);
        private static readonly bool IsMacOS = RuntimeInformation.IsOSPlatform(OSPlatform.OSX);

        private readonly Socket _socket;
        private readonly ISocketsTrace _trace;
        private readonly SocketReceiver _receiver;
        private readonly SocketSender _sender;
        private readonly CancellationTokenSource _connectionClosedTokenSource = new CancellationTokenSource();

        private readonly object _shutdownLock = new object();
        private volatile bool _socketDisposed;
        private volatile Exception _shutdownReason;
        private Task _processingTask;
        private readonly TaskCompletionSource _waitForConnectionClosedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        private bool _connectionClosed;
 
        /// <summary>
        /// 临时暴露
        /// </summary>
        public Socket Socket { get {return _socket; } }

        public SocketConnection(Socket socket,
                                  MemoryPool<byte> memoryPool,
                                  PipeScheduler transportScheduler,
                                  ISocketsTrace trace,
                                  long? maxReadBufferSize = null,
                                  long? maxWriteBufferSize = null,
 
                                  bool useInlineSchedulers = false)
        {
            Debug.Assert(socket != null);
            Debug.Assert(memoryPool != null);
            Debug.Assert(trace != null);

            _socket = socket;
            MemoryPool = memoryPool;
            _trace = trace;
 
            LocalEndPoint = _socket.LocalEndPoint;
            RemoteEndPoint = _socket.RemoteEndPoint;

            ConnectionClosed = _connectionClosedTokenSource.Token;

            // On *nix platforms, Sockets already dispatches to the ThreadPool.
            // Yes, the IOQueues are still used for the PipeSchedulers. This is intentional.
            // https://github.com/aspnet/KestrelHttpServer/issues/2573
            var awaiterScheduler = IsWindows ? transportScheduler : PipeScheduler.Inline;

            var applicationScheduler = PipeScheduler.ThreadPool;
            if (useInlineSchedulers)
            {
                transportScheduler = PipeScheduler.Inline;
                awaiterScheduler = PipeScheduler.Inline;
                applicationScheduler = PipeScheduler.Inline;
            }

            _receiver = new SocketReceiver(_socket, awaiterScheduler);
            _sender = new SocketSender(_socket, awaiterScheduler);

            maxReadBufferSize ??= 0;
            maxWriteBufferSize ??= 0;

            var inputOptions = new PipeOptions(MemoryPool, applicationScheduler, transportScheduler, maxReadBufferSize.Value, maxReadBufferSize.Value / 2, useSynchronizationContext: false);
            var outputOptions = new PipeOptions(MemoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize.Value, maxWriteBufferSize.Value / 2, useSynchronizationContext: false);

            var pair = DuplexPipe.CreateConnectionPair(inputOptions, outputOptions);

            // Set the transport and connection id
            Transport = pair.Transport;
            Application = pair.Application;
        }

        /// <summary>
        /// Transport To Application
        /// </summary>
        public PipeWriter Input => Application.Output;

        /// <summary>
        /// Application To Transport
        /// </summary>
        public PipeReader Output => Application.Input;

        public override MemoryPool<byte> MemoryPool { get; }

 

        public void Start()
        {
            _processingTask = StartAsync();
        }

        private async Task StartAsync()
        {

            try
            {

                // Spawn send and receive logic
                var receiveTask = DoReceive();
                var sendTask = DoSend();

                // Now wait for both to complete
                await receiveTask;
                await sendTask;
                //await Task.WhenAll(receiveTask, sendTask);

                _receiver.Dispose();
                _sender.Dispose();
            }
            catch (Exception ex)
            {
                _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.{nameof(StartAsync)}.");
            }
             
        }

        public override void Abort(ConnectionAbortedException abortReason)
        {
            // Try to gracefully close the socket to match libuv behavior.
            Shutdown(abortReason);

            // Cancel ProcessSends loop after calling shutdown to ensure the correct _shutdownReason gets set.
            Output.CancelPendingRead();
        }

        // Only called after connection middleware is complete which means the ConnectionClosed token has fired.
        public override async ValueTask DisposeAsync()
        {
            Transport.Input.Complete();
            Transport.Output.Complete();

            if (_processingTask != null)
            {
                await _processingTask;
            }

            _connectionClosedTokenSource.Dispose();
        }

        private async Task DoReceive()
        {
            Exception error = null;

            try
            {
                await ProcessReceives();
            }
            catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
            {
                // This could be ignored if _shutdownReason is already set.
                error = new ConnectionResetException(ex.Message, ex);

                // There's still a small chance that both DoReceive() and DoSend() can log the same connection reset.
                // Both logs will have the same ConnectionId. I don't think it's worthwhile to lock just to avoid this.
                if (!_socketDisposed)
                {
                    _trace.ConnectionReset(ConnectionId+" RECEIVE");
                }
            }
            catch (Exception ex)
                when ((ex is SocketException socketEx && IsConnectionAbortError(socketEx.SocketErrorCode)) ||
                       ex is ObjectDisposedException)
            {
                // This exception should always be ignored because _shutdownReason should be set.
                error = ex;

                if (!_socketDisposed)
                {
                    // This is unexpected if the socket hasn't been disposed yet.
                    _trace.ConnectionError(ConnectionId, error);
                }
            }
            catch (Exception ex)
            {
                // This is unexpected.
                error = ex;
                _trace.ConnectionError(ConnectionId, error);
            }
            finally
            {
                // If Shutdown() has already bee called, assume that was the reason ProcessReceives() exited.
                Input.Complete(_shutdownReason ?? error);

                FireConnectionClosed();

                await _waitForConnectionClosedTcs.Task;
            }
        }

        private async Task ProcessReceives()
        {
            // Resolve `input` PipeWriter via the IDuplexPipe interface prior to loop start for performance.
            var input = Input;
            while (true)
            {
 
                // Ensure we have some reasonable amount of buffer space
                var buffer = input.GetMemory(MinAllocBufferSize);

                var bytesReceived = await _receiver.ReceiveAsync(buffer);

                if (bytesReceived == 0)
                {
                    // FIN
                    _trace.ConnectionReadFin(ConnectionId);
                    break;
                }

                input.Advance(bytesReceived);

                var flushTask = input.FlushAsync();

                var paused = !flushTask.IsCompleted;

                if (paused)
                {
                    _trace.ConnectionPause(ConnectionId);
                }

                var result = await flushTask;

                if (paused)
                {
                    _trace.ConnectionResume(ConnectionId);
                }

                if (result.IsCompleted || result.IsCanceled)
                {
                    // Pipe consumer is shut down, do we stop writing
                    break;
                }
            }
        }

        private async Task DoSend()
        {
            Exception shutdownReason = null;
            Exception unexpectedError = null;

            try
            {
                await ProcessSends();
            }
            catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
            {
                shutdownReason = new ConnectionResetException(ex.Message, ex);
                _trace.ConnectionReset(ConnectionId + " SEND");
            }
            catch (Exception ex)
                when ((ex is SocketException socketEx && IsConnectionAbortError(socketEx.SocketErrorCode)) ||
                       ex is ObjectDisposedException)
            {
                // This should always be ignored since Shutdown() must have already been called by Abort().
                shutdownReason = ex;
            }
            catch (Exception ex)
            {
                shutdownReason = ex;
                unexpectedError = ex;
                _trace.ConnectionError(ConnectionId, unexpectedError);
            }
            finally
            {
                Shutdown(shutdownReason);

                // Complete the output after disposing the socket
                Output.Complete(unexpectedError);

                // Cancel any pending flushes so that the input loop is un-paused
                Input.CancelPendingFlush();
            }
        }

        private async Task ProcessSends()
        {
            // Resolve `output` PipeReader via the IDuplexPipe interface prior to loop start for performance.
            var output = Output;
            while (true)
            {
                var result = await output.ReadAsync();

                if (result.IsCanceled)
                {
                    break;
                }

                var buffer = result.Buffer;

                var end = buffer.End;
                var isCompleted = result.IsCompleted;
                if (!buffer.IsEmpty)
                {
                    await _sender.SendAsync(buffer);
                }

                output.AdvanceTo(end);

                if (isCompleted)
                {
                    break;
                }
                if (_connectionClosedTokenSource.IsCancellationRequested)
                {
                    return;
                }
            }
        }
 
        private void FireConnectionClosed()
        {
            // Guard against scheduling this multiple times
            if (_connectionClosed)
            {
                return;
            }

            _connectionClosed = true;

            ThreadPool.UnsafeQueueUserWorkItem(state =>
            {
                state.CancelConnectionClosedToken();

                state._waitForConnectionClosedTcs.TrySetResult();
            },
            this,
            preferLocal: false);
        }

        private void Shutdown(Exception shutdownReason)
        {
            lock (_shutdownLock)
            {
                if (_socketDisposed)
                {
                    return;
                }

                // Make sure to close the connection only after the _aborted flag is set.
                // Without this, the RequestsCanBeAbortedMidRead test will sometimes fail when
                // a BadHttpRequestException is thrown instead of a TaskCanceledException.
                _socketDisposed = true;

                // shutdownReason should only be null if the output was completed gracefully, so no one should ever
                // ever observe the nondescript ConnectionAbortedException except for connection middleware attempting
                // to half close the connection which is currently unsupported.
                _shutdownReason = shutdownReason ?? new ConnectionAbortedException("The Socket transport's send loop completed gracefully.");

                _trace.ConnectionWriteFin(ConnectionId, _shutdownReason.Message);

                try
                {
                    // Try to gracefully close the socket even for aborts to match libuv behavior.
                    _socket.Shutdown(SocketShutdown.Both);
                }
                catch
                {
                    // Ignore any errors from Socket.Shutdown() since we're tearing down the connection anyway.
                }

                _socket.Dispose();
            }
        }

        private void CancelConnectionClosedToken()
        {
            try
            {
                _connectionClosedTokenSource.Cancel();
            }
            catch (Exception ex)
            {
                _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.{nameof(CancelConnectionClosedToken)}.");
            }
        }

        private static bool IsConnectionResetError(SocketError errorCode)
        {
            // A connection reset can be reported as SocketError.ConnectionAborted on Windows.
            // ProtocolType can be removed once https://github.com/dotnet/corefx/issues/31927 is fixed.
            return errorCode == SocketError.ConnectionReset ||
                   errorCode == SocketError.Shutdown ||
                   (errorCode == SocketError.ConnectionAborted && IsWindows) ||
                   (errorCode == SocketError.ProtocolType && IsMacOS);
        }

        private static bool IsConnectionAbortError(SocketError errorCode)
        {
            // Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix.
            return errorCode == SocketError.OperationAborted ||
                   errorCode == SocketError.Interrupted ||
                   (errorCode == SocketError.InvalidArgument && !IsWindows);
        }
    }


    ///// <summary>
    ///// 
    ///// </summary>
    //public class SocketConnection : TransportConnection, IDisposable
    //{
    //    private static readonly bool IsWindows = RuntimeInformation.IsOSPlatform(OSPlatform.Windows);
    //    private static readonly bool IsMacOS = RuntimeInformation.IsOSPlatform(OSPlatform.OSX);

    //    private readonly CancellationTokenSource _connectionClosedTokenSource = new();
    //    private const int MinAllocBufferSize = 2048;
    //    private const int PauseWriterThreshold = 65536; //64k 主动安全附件包默认长度： 62+64K,测试时都是62+4K,808包最小15字节
    //    private const int ResumeWriterTheshold = PauseWriterThreshold / 2;


    //    private readonly object _shutdownLock = new();
    //    private volatile bool _socketDisposed;
    //    private volatile Exception _shutdownReason;

    //    private readonly TaskCompletionSource _waitForConnectionClosedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
    //    private bool _connectionClosed;

    //    private long Counter = 0;
    //    private readonly Socket _socket;
    //    private readonly ISocketsTrace _trace;
    //    private readonly SocketReceiver _receiver;
    //    private readonly SocketSender _sender;

    //    private int _maxWriteBufferSize = 1024;
    //    private long _maxReadBufferSize = 1024;

    //    //public ILogger Logger { get; }
    //    public CancellationToken ConnectionClosed { get; set; }
    //    private Task _processingTask;
    //    private readonly JTActiveSafetySessionManager _sessionManager;
    //    public JTActiveSafetyTcpSession _tcpSession;
    //    private readonly bool _waitForData;
    //    private string _connectionId;
    //    private JT808Serializer _serializer;
    //    private JT808MessageHandler _messageHandler;

    //    public IServiceProvider ServiceProvider { get; private set; }
    //    public ILogger LifetimeLogger { get; private set; }

    //    public readonly IAttachFileStorageService _attachFileStorageService;

    //    /// <summary>
    //    /// 
    //    /// </summary>
    //    /// <param name="serviceProvider"></param>
    //    /// <param name="socket"></param>
    //    /// <param name="sessionManager"></param>
    //    /// <param name="trace"></param>
    //    /// <param name="logger"></param>
    //    /// <param name="maxReadBufferSize"></param>
    //    /// <param name="maxWriteBufferSize"></param>
    //    public SocketConnection(IServiceProvider serviceProvider, Socket socket, JTActiveSafetySessionManager sessionManager,
    //        ISocketsTrace trace, ILogger logger, long? maxReadBufferSize = null, long? maxWriteBufferSize = null, bool waitForData = true)
    //    {
    //        Debug.Assert(socket != null);
    //        Debug.Assert(trace != null);

    //        _socket = socket;
    //        _trace = trace;
    //        _sessionManager = sessionManager;
    //        LifetimeLogger = logger;

    //        var serializer = serviceProvider.GetService<IJT808Config>();
    //        _serializer = serializer.GetSerializer();

    //        _messageHandler = serviceProvider.GetService<JT808MessageHandler>();

    //        _attachFileStorageService = serviceProvider.GetService<IAttachFileStorageService>();

    //        LocalEndPoint = _socket.LocalEndPoint;
    //        RemoteEndPoint = _socket.RemoteEndPoint;
    //        ConnectionClosed = _connectionClosedTokenSource.Token;

    //        if (maxReadBufferSize.HasValue)
    //            _maxReadBufferSize = maxReadBufferSize.Value;
    //        if (maxWriteBufferSize.HasValue)
    //            _maxWriteBufferSize = (int)maxWriteBufferSize.Value;

    //        _waitForData = waitForData;

    //        var transportScheduler = PipeScheduler.ThreadPool; //这里用构造方法参数传入

    //        // On *nix platforms, Sockets already dispatches to the ThreadPool.
    //        // Yes, the IOQueues are still used for the PipeSchedulers. This is intentional.
    //        // https://github.com/aspnet/KestrelHttpServer/issues/2573
    //        var awaiterScheduler = IsWindows ? transportScheduler : PipeScheduler.Inline;
    //        var applicationScheduler = PipeScheduler.ThreadPool;

    //        _receiver = new SocketReceiver(_socket, awaiterScheduler);
    //        _sender = new SocketSender(_socket, awaiterScheduler);

    //        maxReadBufferSize ??= 0;
    //        maxWriteBufferSize ??= 0;

    //        var inputOptions = new PipeOptions(MemoryPool, applicationScheduler, transportScheduler, maxReadBufferSize.Value, maxReadBufferSize.Value / 2, useSynchronizationContext: false);
    //        var outputOptions = new PipeOptions(MemoryPool, transportScheduler, applicationScheduler, maxWriteBufferSize.Value, maxWriteBufferSize.Value / 2, useSynchronizationContext: false);

    //        var pair = DuplexPipe.CreateConnectionPair(inputOptions, outputOptions);

    //        // Set the transport and connection id
    //        Transport = pair.Transport;
    //        Application = pair.Application;
    //    }

    //    public PipeWriter Input => Application.Output;

    //    public PipeReader Output => Application.Input;

    //    /// <summary>
    //    /// 
    //    /// </summary>
    //    public void Start()
    //    {
    //        _processingTask = StartAsync();
    //    }

    //    private async Task StartAsync(CancellationToken cancellationToken = default)
    //    {
    //        var connectionManager = ServiceLocator.Instance.GetRequiredService<ConnectionManager>();
    //        //connectionManager.Add(this);
    //        try
    //        {
    //            // session and connection One-to-one
    //            _tcpSession = new JTActiveSafetyTcpSession(_socket);
    //            _sessionManager.TryAdd(_tcpSession);

    //            var receiveTask = DoReceive();
    //            var sendTask = DoSend();

    //            // Now wait for both to complete

    //            await Task.WhenAll(receiveTask, sendTask);

    //            _receiver.Dispose();

    //            _sessionManager.RemoveBySessionId(_tcpSession.SessionID);
    //        }
    //        catch (Exception ex)
    //        {
    //            _trace.LogError(0, ex, $"Unexpected exception in {nameof(ConnectionId)}.{nameof(StartAsync)}.");
    //        }
    //        finally
    //        {
    //            connectionManager.Remove(this.ConnectionId);
    //        }
    //        //await Task.Factory.StartNew(async () =>
    //        //{ }, cancellationToken);
    //    }

    //    public override void Abort(ConnectionAbortedException abortReason)
    //    {
    //        // Try to gracefully close the socket to match libuv behavior.
    //        Shutdown(abortReason);

    //        // Cancel ProcessSends loop after calling shutdown to ensure the correct _shutdownReason gets set.
    //        Output.CancelPendingRead();
    //    }

    //    // Only called after connection middleware is complete which means the ConnectionClosed token has fired.
    //    public override async ValueTask DisposeAsync()
    //    {
    //        Transport.Input.Complete();
    //        Transport.Output.Complete();

    //        if (_processingTask != null)
    //        {
    //            await _processingTask;
    //        }

    //        _connectionClosedTokenSource.Dispose();
    //    }

    //    //private async Task NewSession(CancellationToken token = default)
    //    //{
    //    //    //PauseWriterThreshold：确定在调用 FlushAsync 暂停之前应缓冲多少数据。
    //    //    //ResumeWriterThreshold：确定在恢复对 PipeWriter.FlushAsync 的调用之前，读取器必须观察多少数据。
    //    //    //当Pipe的数据量超过PauseWriterThreshold，PipeWriter.FlushAsync会异步阻塞。数据量变得低于ResumeWriterThreshold，它会解锁时。两个值用于防止在极限附近发生反复阻塞和解锁。

    //    //    //var options = new PipeOptions(pauseWriterThreshold: 10/*暂停写入阈值*/, resumeWriterThreshold: 5/*恢复写入阈值*/,
    //    //    //    readerScheduler: PipeScheduler.ThreadPool,
    //    //    //    writerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
    //    //    //,
    //    //    //    readerScheduler: PipeScheduler.Inline,
    //    //    //    writerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false
    //    //    var options = new PipeOptions(pauseWriterThreshold: PauseWriterThreshold/*暂停写入阈值*/, resumeWriterThreshold: ResumeWriterTheshold/*恢复写入阈值*/);
    //    //   var pipe = new Pipe(options);
    //    //    Task writing = FillPipeAsync(pipe.Writer, token);
    //    //    Task reading = ReadPipeAsync(pipe.Reader, token);
    //    //    await Task.WhenAll(reading, writing);
    //    //}
    //    private async Task DoReceive()
    //    {
    //        Exception error = null;

    //        try
    //        {
    //            await ProcessReceives();
    //        }
    //        catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
    //        {
    //            // This could be ignored if _shutdownReason is already set.
    //            error = new ConnectionResetException(ex.Message, ex);

    //            // There's still a small chance that both DoReceive() and DoSend() can log the same connection reset.
    //            // Both logs will have the same ConnectionId. I don't think it's worthwhile to lock just to avoid this.
    //            if (!_socketDisposed)
    //            {
    //                _trace.ConnectionReset(ConnectionId);
    //            }
    //        }
    //        catch (Exception ex)
    //            when ((ex is SocketException socketEx && IsConnectionAbortError(socketEx.SocketErrorCode)) ||
    //                   ex is ObjectDisposedException)
    //        {
    //            // This exception should always be ignored because _shutdownReason should be set.
    //            error = ex;

    //            if (!_socketDisposed)
    //            {
    //                // This is unexpected if the socket hasn't been disposed yet.
    //                _trace.ConnectionError(ConnectionId, error);
    //            }
    //        }
    //        catch (Exception ex)
    //        {
    //            // This is unexpected.
    //            error = ex;
    //            _trace.ConnectionError(ConnectionId, error);
    //        }
    //        finally
    //        {
    //            Input.Complete(_shutdownReason ?? error);
    //        }
    //    }

    //    private async Task ProcessReceives()
    //    {
    //        // Resolve `input` PipeWriter via the IDuplexPipe interface prior to loop start for performance.
    //        var input = Input;
    //        while (true)
    //        {
    //            //if (_waitForData)
    //            //{
    //            //    // Wait for data before allocating a buffer.
    //            //    await _receiver.WaitForDataAsync();
    //            //}
    //            // MacOS blocked on https://github.com/dotnet/corefx/issues/31766
    //            if (!IsMacOS)
    //            {
    //                // Wait for data before allocating a buffer.
    //                await _receiver.WaitForDataAsync();
    //            }

    //            // Ensure we have some reasonable amount of buffer space
    //            var buffer = input.GetMemory(MinAllocBufferSize);

    //            var bytesReceived = await _receiver.ReceiveAsync(buffer);
    //            //var bytesReceived = await _socket.ReceiveAsync(buffer, SocketFlags.None);
    //            if (bytesReceived == 0)
    //            {
    //                // FIN
    //                _trace.ConnectionReadFin(ConnectionId);
    //                break;
    //            }

    //            input.Advance(bytesReceived);

    //            var flushTask = input.FlushAsync();

    //            var paused = !flushTask.IsCompleted;

    //            if (paused)
    //            {
    //                _trace.ConnectionPause(ConnectionId);
    //            }

    //            var result = await flushTask;

    //            if (paused)
    //            {
    //                _trace.ConnectionResume(ConnectionId);
    //            }

    //            if (result.IsCompleted || result.IsCanceled)
    //            {
    //                // Pipe consumer is shut down, do we stop writing
    //                break;
    //            }
    //        }
    //    }

    //    private async Task DoSend()
    //    {
    //        Exception shutdownReason = null;
    //        Exception unexpectedError = null;

    //        try
    //        {
    //            await ProcessSends();
    //        }
    //        catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
    //        {
    //            shutdownReason = new ConnectionResetException(ex.Message, ex); ;
    //            _trace.ConnectionReset(ConnectionId);
    //        }
    //        catch (Exception ex)
    //            when ((ex is SocketException socketEx && IsConnectionAbortError(socketEx.SocketErrorCode)) ||
    //                   ex is ObjectDisposedException)
    //        {
    //            // This should always be ignored since Shutdown() must have already been called by Abort().
    //            shutdownReason = ex;
    //        }
    //        catch (Exception ex)
    //        {
    //            shutdownReason = ex;
    //            unexpectedError = ex;
    //            _trace.ConnectionError(ConnectionId, unexpectedError);
    //        }
    //        finally
    //        {
    //            Shutdown(shutdownReason);

    //            // Complete the output after disposing the socket
    //            Output.Complete(unexpectedError);

    //            // Cancel any pending flushes so that the input loop is un-paused
    //            Input.CancelPendingFlush();
    //        }
    //    }

    //    private async Task ProcessSends()
    //    {
    //        // Resolve `output` PipeReader via the IDuplexPipe interface prior to loop start for performance.
    //        var output = Output;
    //        while (true)
    //        {
    //            if (_connectionClosedTokenSource.IsCancellationRequested)
    //            {
    //                return;
    //            }

    //            var result = await output.ReadAsync();//注意在调用 PipeReader.ReadAsync 之后始终调用 PipeReader.AdvanceTo

    //            var buffer = result.Buffer;
    //            var consumed = buffer.Start;
    //            var examined = buffer.End;

    //            try
    //            {
    //                if (result.IsCanceled)
    //                {
    //                    break;//读取已取消。您可以不读取现有数据而退出。
    //                }

    //                if (!buffer.IsEmpty)
    //                {
    //                    var r = ReadBuffer(ref buffer, ref consumed, ref examined);
    //                    if (r != null)
    //                    {
    //                        await _sender.SendAsync(buffer);
    //                    }
    //                }

    //                if (result.IsCompleted)
    //                {
    //                    //if (buffer.Length > 0)
    //                    //{
    //                    //    // The message is incomplete and there's no more data to process.
    //                    //    throw new InvalidDataException("Incomplete message.");
    //                    //}
    //                    break;//没有更多的数据要处理
    //                }
    //            }
    //            catch (Exception ex)
    //            {

    //                _trace.LogError(ex, $"[ReadPipe Error]:{_tcpSession.Client.RemoteEndPoint}");
    //                break;
    //            }
    //            finally
    //            {
    //                output.AdvanceTo(consumed);//已读取了多少
    //                                           //reader.AdvanceTo(buffer.GetPosition(index));
    //                                           //reader.AdvanceTo(buffer.Start, buffer.End);
    //                                           //reader.AdvanceTo(consumed/*确定消耗的内存量*/, examined/*确定观察到的缓冲区数*/);
    //            }
    //        }
    //    }

    //    //private async Task FillPipeAsync(PipeWriter writer, CancellationToken token = default)
    //    //{
    //    //    Exception shutdownReason = null;
    //    //    Exception unexpectedError = null;
    //    //     Resolve `input` PipeWriter via the IDuplexPipe interface prior to loop start for performance.
    //    //    var input = Input;
    //    //    try
    //    //    {
    //    //        while (!_connectionClosedTokenSource.IsCancellationRequested)
    //    //        {
    //    //            if (token.IsCancellationRequested)
    //    //            {
    //    //                return;
    //    //            }

    //    //            try
    //    //            {
    //    //                Memory<byte> memory = writer.GetMemory(_maxWriteBufferSize);

    //    //                设备多久没发数据就断开连接 Receive Timeout.
    //    //                int bytesRead = await _receiver.ReceiveAsync(memory, SocketFlags.None);
    //    //                if (bytesRead == 0)
    //    //                {
    //    //                    break;
    //    //                }
    //    //                writer.Advance(bytesRead);// 告诉PipeWriter从套接字读取了多少
    //    //            }
    //    //            catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
    //    //            {
    //    //                shutdownReason = new ConnectionResetException(ex.Message, ex);
    //    //                _trace.ConnectionReset(ConnectionId);
    //    //            }
    //    //            catch (Exception ex)
    //    //                when ((ex is SocketException socketEx && IsConnectionAbortError(socketEx.SocketErrorCode)) ||
    //    //                       ex is ObjectDisposedException)
    //    //            {
    //    //                 This should always be ignored since Shutdown() must have already been called by Abort().
    //    //                shutdownReason = ex;
    //    //            }
    //    //            catch (Exception ex)
    //    //            {
    //    //                shutdownReason = ex;
    //    //                unexpectedError = ex;
    //    //                _trace.ConnectionError(ConnectionId, unexpectedError);
    //    //            }

    //    //            FlushResult result = await writer.FlushAsync(token);// 标记数据可用，让PipeReader读取
    //    //            if (result.IsCompleted)
    //    //            {
    //    //                break;
    //    //            }
    //    //        }
    //    //    }
    //    //    finally
    //    //    {
    //    //        await writer.CompleteAsync();// 告诉PipeReader没有更多的数据

    //    //        Shutdown(shutdownReason);

    //    //         Complete the output after disposing the socket
    //    //        Output.Complete(unexpectedError);

    //    //         Cancel any pending flushes so that the input loop is un-paused
    //    //        Input.CancelPendingFlush();
    //    //    }
    //    //}
    //    //private async Task ReadPipeAsync()
    //    //{
    //    //    try
    //    //    {
    //    //        while (true)
    //    //        {
    //    //            if (_connectionClosedTokenSource.IsCancellationRequested)
    //    //            {
    //    //                return;
    //    //            }

    //    //            var result = await Output.ReadAsync();//注意在调用 PipeReader.ReadAsync 之后始终调用 PipeReader.AdvanceTo

    //    //            var buffer = result.Buffer;
    //    //            var consumed = buffer.Start;
    //    //            var examined = buffer.End;

    //    //            try
    //    //            {
    //    //                if (result.IsCanceled)
    //    //                {
    //    //                    break;//读取已取消。您可以不读取现有数据而退出。
    //    //                }

    //    //                if (!buffer.IsEmpty)
    //    //                {
    //    //                    ReadBuffer(ref buffer, ref consumed, ref examined);
    //    //                }

    //    //                if (result.IsCompleted)
    //    //                {
    //    //                    //if (buffer.Length > 0)
    //    //                    //{
    //    //                    //    // The message is incomplete and there's no more data to process.
    //    //                    //    throw new InvalidDataException("Incomplete message.");
    //    //                    //}
    //    //                    break;//没有更多的数据要处理
    //    //                }
    //    //            }
    //    //            catch (Exception ex)
    //    //            {

    //    //                _trace.LogError(ex, $"[ReadPipe Error]:{_tcpSession.Client.RemoteEndPoint}");
    //    //                break;
    //    //            }
    //    //            finally
    //    //            {
    //    //                Output.AdvanceTo(consumed);//已读取了多少
    //    //                //reader.AdvanceTo(buffer.GetPosition(index));
    //    //                //reader.AdvanceTo(buffer.Start, buffer.End);
    //    //                //reader.AdvanceTo(consumed/*确定消耗的内存量*/, examined/*确定观察到的缓冲区数*/);
    //    //            }
    //    //        }
    //    //    }
    //    //    finally
    //    //    {

    //    //        await Output.CompleteAsync();//.Complete();
    //    //    }
    //    //}

    //    /// <summary>
    //    /// 读取包消息，每次只读取一条808或主动安全附件消息。
    //    /// </summary>
    //    /// <param name="buffer">源</param>
    //    /// <param name="consumed">读取进度</param>
    //    /// <param name="examined"></param>
    //    /// <returns></returns>
    //    private byte[] ReadBuffer(ref ReadOnlySequence<byte> buffer, ref SequencePosition consumed, ref SequencePosition examined)
    //    {
    //        //读取一个完整包 常见异常及处理方式
    //        //1.长度不够   终止打包，等待下次读取。
    //        //2.已找到808或主动安全包头，读取时未找到终止符(808)或长度不够（主动安全）  终止打包，等待下次读取。
    //        //3.找不到808或主动安全包头，终止打包，等待下次读取。
    //        byte[] r = null;
    //        var reader = new SequenceReader<byte>(buffer);
    //        while (!reader.End)
    //        {
    //            //808包最小14位 主动安全附件文件包最小62位
    //            if (reader.Length < 14)
    //            {
    //                break;
    //            }

    //            var x = ExtractActiveSafetyPack(ref reader, _trace);
    //            if (x == 808)
    //            {
    //                if (reader.TryRead808Package(out var _span))
    //                {

    //                    _trace.LogDebug($"[808 Parse ===>]:{_span.ToArray().ToHexString()},{_tcpSession.Client.RemoteEndPoint},{_tcpSession.TerminalPhoneNo}");

    //                    r = Hander808Parse(_span.ToArray());

    //                    break;
    //                }

    //            }
    //            else if (x == 999)
    //            {
    //                if (reader.TryReadActiveSafetyPackage(out var _span2))
    //                {

    //                    //_trace.LogDebug($"[ActiveSafety Parse ===>]:{_tcpSession.Client.RemoteEndPoint},{_tcpSession.TerminalPhoneNo}");
    //                    _trace.LogDebug($"[ActiveSafety Parse ===>]:{_span2.ToArray().ToHexString()},{_tcpSession.Client.RemoteEndPoint},{_tcpSession.TerminalPhoneNo}");

    //                    HanderActiveSafetyParse(_span2.ToArray());

    //                    break;
    //                }
    //            }
    //            else
    //            {
    //                break;
    //            }
    //        }
    //        if (reader.End)
    //        {
    //            examined = consumed = buffer.End;
    //        }
    //        else
    //        {
    //            consumed = reader.Position;
    //        }
    //        return r;
    //    }
    //    /// <summary>
    //    /// 从数据中提取出符合主动安全规范的包（包括3个交互808指令0x1210/0x1211/0x1212 和上传的主动安全附件文件数据包），不符合的数据将被截取并丢弃。
    //    /// 同时，这也是一个安全检查，服务器对外提供服务不可避免的遇到网络刺探。
    //    /// </summary>
    //    /// <param name="reader"></param>
    //    /// <param name="trace"></param>
    //    private static int ExtractActiveSafetyPack(ref SequenceReader<byte> reader, ISocketsTrace trace)
    //    {
    //        if (reader.TryReadBigEndian(out short sss))
    //            reader.Rewind(2);

    //        if (sss == 32274 /*0x7e 0x12*/)
    //        {
    //            return 808;
    //        }

    //        if (reader.TryReadBigEndian(out int ss))
    //            reader.Rewind(4);
    //        if (ss == JTActiveSafetyPackage.FH)
    //        {
    //            return 999;
    //        }

    //        trace.LogWarning($"收到非808或主动安全包头：{reader.Sequence.ToArray().ToHexString()}");
    //        if (reader.TryReadTo(out ReadOnlySpan<byte> _span, new byte[] { 0x7E, 0x12 }, false))
    //        {
    //            trace.LogInformation($"已剔除数据:{_span.ToArray().ToHexString()}");
    //            return 0;
    //        }
    //        //if (reader.TryReadTo(out ReadOnlySpan<byte> _, new byte[] { 0x7E, 0x92 }, false))
    //        //{
    //        //    trace.LogInformation($"已剔除数据:{_span.ToArray().ToHexString()}");
    //        //    return 0;
    //        //}
    //        if (reader.TryReadTo(out ReadOnlySpan<byte> _, new byte[] { 0x30, 0x31, 0x63, 0x64 }, false))
    //        {
    //            trace.LogInformation($"已剔除数据:{_span.ToArray().ToHexString()}");
    //            return 0;
    //        }

    //        reader.Advance(reader.Remaining);
    //        return 0;
    //    }

    //    private byte[] Hander808Parse(byte[] span)
    //    {
    //        try
    //        {

    //            var package = _serializer.HeaderDeserialize(span);
    //            _sessionManager.TryLink(package.Header.TerminalPhoneNo, _tcpSession);
    //            return _messageHandler.Processor(package);
    //        }
    //        catch (JT808Exception ex)
    //        {
    //            LifetimeLogger.LogError($"[HeaderDeserialize ErrorCode1]:{ ex.ErrorCode}," +
    //                $"[ReaderBuffer]:{span.ToHexString()},{_tcpSession.Client.RemoteEndPoint},{_tcpSession.TerminalPhoneNo}");
    //        }
    //        catch (Exception ex)
    //        {
    //            _trace.ConnectionError(ConnectionId, ex);
    //        }
    //        return null;
    //    }

    //    private void HanderActiveSafetyParse(byte[] span)
    //    {
    //        if (_tcpSession?.TerminalPhoneNo == default)
    //        {
    //            LifetimeLogger.LogInformation($"处理主动安全包时找不到会话终端手机号");
    //            return;
    //        }
    //        try
    //        {
    //            _sessionManager.TryLink(_tcpSession.TerminalPhoneNo, _tcpSession);
    //            _attachFileStorageService.WriteData(_tcpSession.TerminalPhoneNo, span).ConfigureAwait(false);
    //            //await _msgProducer.WriteData(session.TerminalPhoneNo, span);

    //            //var _span = ((ushort)0x1299).CreateCustomMsgId(session.TerminalPhoneNo, new JT808_0x1299()
    //            //{
    //            //    FileName = Encoding.UTF8.GetString(span.Slice(4, 50).ToArray()),
    //            //    Offset = BinaryPrimitives.ReadUInt32BigEndian(span.Slice(54, 4)),
    //            //    Length = BinaryPrimitives.ReadUInt32BigEndian(span.Slice(58, 4)),
    //            //});
    //            //var jt808_1299 = _serializer.Serialize(_span).ToArray();
    //            //var package = _serializer.HeaderDeserialize(jt808_1299);
    //            //var downData = _messageHandler.Processor(package);
    //            //session.SendAsync(downData);
    //        }
    //        catch (Exception ex)
    //        {
    //            _trace.ConnectionError(ConnectionId, ex);
    //        }
    //    }

    //    private void FireConnectionClosed()
    //    {
    //        // Guard against scheduling this multiple times
    //        if (_connectionClosed)
    //        {
    //            return;
    //        }

    //        _connectionClosed = true;

    //        ThreadPool.UnsafeQueueUserWorkItem(state =>
    //        {
    //            state.CancelConnectionClosedToken();

    //            state._waitForConnectionClosedTcs.TrySetResult();
    //        },
    //        this,
    //        preferLocal: false);
    //    }
    //    private void Shutdown(Exception shutdownReason)
    //    {
    //        lock (_shutdownLock)
    //        {
    //            if (_socketDisposed)
    //            {
    //                return;
    //            }

    //            // Make sure to close the connection only after the _aborted flag is set.
    //            // Without this, the RequestsCanBeAbortedMidRead test will sometimes fail when
    //            // a BadHttpRequestException is thrown instead of a TaskCanceledException.
    //            _socketDisposed = true;

    //            // shutdownReason should only be null if the output was completed gracefully, so no one should ever
    //            // ever observe the nondescript ConnectionAbortedException except for connection middleware attempting
    //            // to half close the connection which is currently unsupported.
    //            _shutdownReason = shutdownReason ?? new ConnectionAbortedException("The Socket transport's send loop completed gracefully.");

    //            _trace.ConnectionWriteFin(ConnectionId, _shutdownReason.Message);

    //            try
    //            {
    //                // Try to gracefully close the socket even for aborts to match libuv behavior.
    //                _socket.Shutdown(SocketShutdown.Both);
    //            }
    //            catch
    //            {
    //                // Ignore any errors from Socket.Shutdown() since we're tearing down the connection anyway.
    //            }

    //            _socket.Dispose();
    //        }
    //    }

    //    private void CancelConnectionClosedToken()
    //    {
    //        try
    //        {
    //            _connectionClosedTokenSource.Cancel();
    //        }
    //        catch (Exception ex)
    //        {
    //            _trace.LogError(0, ex, $"Unexpected exception in {nameof(Internal.SocketConnection)}.{nameof(CancelConnectionClosedToken)}.");
    //        }
    //    }
    //    private static bool IsConnectionResetError(SocketError errorCode)
    //    {
    //        // A connection reset can be reported as SocketError.ConnectionAborted on Windows.
    //        // ProtocolType can be removed once https://github.com/dotnet/corefx/issues/31927 is fixed.
    //        return errorCode == SocketError.ConnectionReset ||
    //               errorCode == SocketError.Shutdown ||
    //               (errorCode == SocketError.ConnectionAborted && IsWindows) ||
    //               (errorCode == SocketError.ProtocolType && IsMacOS);
    //    }

    //    private static bool IsConnectionAbortError(SocketError errorCode)
    //    {
    //        // Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix.
    //        return errorCode == SocketError.OperationAborted ||
    //               errorCode == SocketError.Interrupted ||
    //               (errorCode == SocketError.InvalidArgument && !IsWindows);
    //    }

    //    #region IDisposable
    //    private bool disposedValue;


    //    protected virtual void Dispose(bool disposing)
    //    {
    //        if (!disposedValue)
    //        {
    //            if (disposing)
    //            {
    //                // TODO: 释放托管状态(托管对象)
    //            }

    //            // TODO: 释放未托管的资源(未托管的对象)并替代终结器
    //            // TODO: 将大型字段设置为 null
    //            disposedValue = true;
    //        }
    //    }

    //    // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
    //    // ~JTActiveSafetyPipeline()
    //    // {
    //    //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //    //     Dispose(disposing: false);
    //    // }

    //    public void Dispose()
    //    {
    //        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //        Dispose(disposing: true);
    //        GC.SuppressFinalize(this);
    //    }
    //    #endregion
    //}



}